package com.ustcinfo.study.mr.pq;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 手机端访问与浏览器访问的占比
 * @author pengqaing
 */
public class PcAndPhone {
    private static class MyMapper extends Mapper<Object, Text, IntWritable,Text>{
        private static final IntWritable count = new IntWritable(1);
        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] arr = line.split("-"); //分隔符
            if (arr[0].trim().length() > 0) {
                if(arr[0].trim().equals("172.16.1.1"))
                    context.write(count,new Text("1-0"));  //定义输入value为 x-x 形式，前一个x为phone数量，后一个为pc数量
                else
                    context.write(count,new Text("0-1"));  //定义同一个key，方便同一个reduce做统计
            }
        }
    }
    private static class MyCombiner extends Reducer<IntWritable,Text,IntWritable,Text>{  //局部reduce优化
        @Override
        protected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            int pc = 0;
            int phone = 0;
            for (Text val : values){
                String[] arr = val.toString().split("-");
                phone += Integer.parseInt(arr[0]);        //在map输出的结果下做局部统计
                pc += Integer.parseInt(arr[1]);           //局部加和，减少Map输出。
            }
            context.write(key,new Text(phone+"-"+pc)); //每一个combiner输出即为 x（phone） - x （pc），key为输入key
        }
    }
    private static class MyReduce extends Reducer<IntWritable, Text, Text, Text>{
        @Override
        protected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            double phone = 0;
            double phonei = 0;
            double pc = 0;
            double pci = 0;
            for(Text val :values){
                String[] arr = val.toString().split("-");  //reduce总加和
                phone += Integer.parseInt(arr[0]);
                pc += Integer.parseInt(arr[1]);
            }
            phonei = phone /(phone+pc);  //获得phone在总量中的占比
            pci = 1 - phonei ;      //或者pc占比 1 - phone占比
            context.write(new Text("phone"),new Text(phonei * 100+"%"));  //输出phone占比
            context.write(new Text("pc"),new Text(pci * 100+"%"));      //输出pc占比
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance();
        job.setJobName("pq_pc_phone");
        job.setJarByClass(PcAndPhone.class);

        job.setMapperClass(MyMapper.class);  //map类
        job.setCombinerClass(MyCombiner.class); //combiner类
        job.setReducerClass(MyReduce.class);  //reduce类

        job.setMapOutputKeyClass(IntWritable.class); //取消map默认输出格式
        job.setMapOutputValueClass(Text.class);     //自定义map输出格式

        job.setOutputKeyClass(Text.class);    //reduce输出格式
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job,new Path(args[0]));  //输入文件路径
        FileOutputFormat.setOutputPath(job,new Path(args[1])); //输出文件路径

        int status = job.waitForCompletion(true) ? 0 : 1;
        System.exit(status);
    }
}
